package me.tuyou.stream.window;

import me.tuyou.common.MyLog;
import me.tuyou.common.MyLogGenerator;
import org.apache.flink.api.common.eventtime.WatermarkStrategy;
import org.apache.flink.api.common.typeinfo.Types;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;

import java.time.Duration;

/**
 * @author tuyou_cd
 * @date 2021/6/24 15:22
 */
public class EventTimeTest {

    public static void main(String[] args) throws Exception {
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();

        WatermarkStrategy<MyLog> watermarkStrategy = WatermarkStrategy.
                <MyLog>forBoundedOutOfOrderness(Duration.ofSeconds(10))
                .withTimestampAssigner((event, timestamp) -> event.getTime().getTime());
        SingleOutputStreamOperator<Tuple2> dataStream = env.addSource(new MyLogGenerator())
                .assignTimestampsAndWatermarks(watermarkStrategy)
                .map(item -> new Tuple2("设备：" + item.getDeviceId(), 1))
                .returns(Types.TUPLE(Types.STRING, Types.INT))
                .keyBy(0)
                .sum(1);

        dataStream.print();
        env.execute();
    }
}
